package cn.simon.free.redis.mq.core.consumer;

import cn.simon.free.redis.mq.enums.ConsumerType;
import cn.simon.free.redis.mq.model.Message;
import cn.simon.free.redis.mq.model.delay.DelayMessage;
import cn.simon.free.redis.mq.model.delay.DelayMessageRo;

import java.util.Objects;
import java.util.Optional;

/**
 * 延迟队列
 *
 * @author simon
 * @date 2021/7/1 16:38
 * @copyright:
 * @modifier
 * @motto a no talent, people who have to study hard
 */
public abstract class DelayRedisMessageConsumer<T> extends AbstractRedisMessageConsumer<T> {

    @Override
    public final Message getMessage(String queueName, ConsumerType consumerType) {
        DelayMessage message = (DelayMessage) redisMessageSupport.isClusterAndGetMessage(queueName, consumerType);
        if (Optional.ofNullable(message).isPresent()) {
            if (message.getDelayTime() < System.currentTimeMillis()) {
                DelayMessageRo delayMessageRo = redisMessageDelayRoSupport.get(message.getTaskId());
                //对象message不存在,直接删除延迟队列中的数据
                if (Objects.isNull(delayMessageRo)) {
                    redisMessageSupport.delete(queueName, message);
                    return null;
                }
                message.setMessage(delayMessageRo.getMessage());
                return message;
            }
            //正常情况,未到时间时,需要重新放回队列,只有集群的时候,才会触发该操作,因为集群情况下,取值采用的是poll模式.
            rePushMessage(queueName, message, consumerType);
            return null;
        }
        return null;
    }

    /**
     * 重新放回队列消费
     * 当类型为集群模式{@link ConsumerType.CLUSTER}时,需要重新放回队列.
     * 1.集群采用的poll模式
     * 2.广播采用的get模式
     *
     * @param queueName
     * @param message
     * @param consumerType
     */
    @Override
    public final void rePushMessage(String queueName, Message message, ConsumerType consumerType) {
        if (redisMessageSupport.isCluster(consumerType)) {
            redisMessageSupport.push(queueName, message, ((DelayMessage) message).getDelayTime());
        }
    }

    @Override
    public final void deleteMessage(String queueName, Message message) {
        redisMessageSupport.delete(queueName, message);
        redisMessageDelayRoSupport.deleteByExpire(message.getTaskId());
    }
}
